package com.aliyun.adb.contest;

import com.aliyun.adb.contest.spi.AnalyticDB;

import java.io.*;
import java.util.*;
import java.util.stream.Collectors;

public class AnalyticDBImpl implements AnalyticDB {
    @Override
    public void load(String tpchDataFileDir, String workspaceDir) throws Exception {
        File dir = new File(tpchDataFileDir);
        for (File dataFile : dir.listFiles()) {
            System.out.println("Start loading table " + dataFile.getName());

//            // You can write data to workspaceDir
//            File yourDataFile = new File(workspaceDir, dataFile.getName());
//            yourDataFile.createNewFile();
//            loadInMemroy(dataFile);
            long start = System.currentTimeMillis();
            loadFromFile(dataFile,workspaceDir);
            System.out.println("Cost: "+(System.currentTimeMillis() - start));

        }
    }

    private void mergeFile(String workspaceDir,String rsltFilePath,String[] columns,int fileCount) throws IOException {
        // 采用选择置换排序
        BufferedWriter bw;
        ArrayList initValues = new ArrayList<Long>();
        ArrayList<BufferedReader> brArr;

        for(int ci = 0;ci < columns.length;ci++){
            brArr = new ArrayList<>();
            bw= new BufferedWriter(new FileWriter(new File(rsltFilePath+columns[ci])));
            for(int i=0;i<fileCount;i++){
                BufferedReader br = new BufferedReader(new FileReader(new File(workspaceDir+File.separator+columns[ci]+i)));
                initValues.add(Long.parseLong(br.readLine()));
                brArr.add(br);
            }
            LoserTree loserTree = new LoserTree(initValues);
            Integer win;
            BufferedReader tempBr = null;
            String tempLine = null;
            while(brArr.size()!=0){
                win= loserTree.getWinner();
                // 写入文件的数据
                Long realData = (Long)loserTree.getLeaf(win);
//                System.out.println(columns[ci]+":"+realData);
                bw.write(realData+"");
                bw.newLine();
                // 根据win来选择下一个
                if((tempBr = brArr.get(win))!=null) {
                    if((tempLine = tempBr.readLine())!=null){
                        // 加入到败者树中
                        loserTree.add(Long.parseLong(tempLine),win);
                    }else{
                        // 已经读到文件末尾，从brArr中移除
                        brArr.remove(tempBr);
                        loserTree.del(win);
                    }
                }
            }

            bw.flush();
            bw.close();
        }

    }

     private void loadFromFile(File inputFile,String workspaceDir) throws IOException {
        BufferedReader reader = new BufferedReader(new FileReader(inputFile));
        String table = inputFile.getName();
        String[] columns = reader.readLine().split(",");

        String rawRow;
        Map<String,Long[]> tempMap = new HashMap<>();
        int maxLen = 1000000;
        int tempIndex = 0;
        int fileIndex = 0;
        for (int i = 0; i < columns.length; i++) {
            tempMap.put(columns[i], new Long[maxLen]);
        }
        long start1 = System.currentTimeMillis();

        // 读到内存中
        while ((rawRow = reader.readLine()) != null) {
            String[] row = rawRow.split(",");
            for (int i = 0; i < columns.length; i++) {
                tempMap.get(columns[i])[tempIndex] = Long.parseLong( row[i]);
            }
            tempIndex++;
            if(tempIndex == maxLen){
                // 排序，并写入文件
                sortAndWriteToFile(columns,tempMap,workspaceDir,fileIndex);
                // 再次操作
                tempIndex = 0;
                for(int i = 0; i < columns.length; i++) {
                    tempMap.put (columns[i], new Long[maxLen]);
                }
                fileIndex++;
            }
        }

        // 如果有剩余，将剩余的写入文件
          if(tempIndex!=0){

              // 去除空数组
              for(int i = 0; i < columns.length; i++) {
                  Long[] temp = new Long[tempIndex];
                  System.arraycopy(tempMap.get(columns[i]),0,temp,0, tempIndex);
                  tempMap.put (columns[i], temp);
              }

              // 排序，并写入文件
              sortAndWriteToFile(columns,tempMap,workspaceDir,fileIndex);
          }

         System.out.println("Read cost: " + (System.currentTimeMillis()-start1));
          long start2 = System.currentTimeMillis();
        // 利用置换选择排序合并文件
         String rsltFilePath = workspaceDir + File.separator +"_result";

         mergeFile(workspaceDir,rsltFilePath,columns,fileIndex);

        // 结束
         System.out.println("Merge cost: " + (System.currentTimeMillis()-start2));
    }

    private void sortAndWriteToFile(String[] columns, Map<String,Long[]> dataMap,String workspaceDir,int fileIndex) throws IOException{
        for (int i = 0; i < columns.length; i++) {

            Long[] tempData = dataMap.get(columns[i]);
            if(tempData==null || tempData.length==0){
                break;
            }
            Arrays.sort(tempData);
            File outFile = new File(workspaceDir,columns[i]+fileIndex);

            if(outFile.exists()){
                if(!outFile.delete()){
                  throw new IOException("文件删除失败！");
                }
            }
            if(!outFile.createNewFile()){
                throw new IOException("创建文件失败");
            }
            BufferedWriter br = new BufferedWriter(new FileWriter(outFile));
            Long[] longs = dataMap.get(columns[i]);
            String s= Arrays.stream(longs).map(Objects::toString).collect(Collectors.joining("\n"));
            br.write(s);
            br.flush();
            br.close();
        }
    }


    @Override
    public String quantile(String table, String column, double percentile) throws Exception {
        return null;
    }
}
